package cn.com.greatwall.kafka.service.impl;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;

import org.springframework.security.core.userdetails.UserDetails;
import org.springframework.stereotype.Service;

import cn.com.greatwall.common.utils.PageUtil;
import cn.com.greatwall.kafka.service.KafkaService;
import cn.com.greatwall.kafka.service.MonitorService;
import cn.com.greatwall.kafka.service.dto.ClusterDto;
import cn.com.greatwall.kafka.service.dto.MonitorTopicDto;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;

/**
 * @Author herw
 * @Time 2021-05-10 15:17:25
 * @Version 1.0
 * @Description: TODO(用一句话描述该文件做什么)
 */
@Slf4j
@Service(value = "mq-monitorServiceImpl")
@RequiredArgsConstructor
public class MonitorServiceImpl implements MonitorService {
    private final KafkaService kafkaService;

    @Override
    public Map<String, Object> getTopicList(List<ClusterDto> clusterDtos, UserDetails user) {
        List<MonitorTopicDto> monitorTopicDtos = new ArrayList<>();
        for (ClusterDto clusterDto : clusterDtos) {
            try {
                Set<String> topicMap = kafkaService.getKafkaUtil(clusterDto.getId().toString()).listTopics();
                topicMap.forEach(topic -> {
                    MonitorTopicDto monitorTopicDto = new MonitorTopicDto();
                    monitorTopicDto.setClusterId(clusterDto.getId());
                    monitorTopicDto.setClusterName(clusterDto.getName());
                    monitorTopicDto.setTopicName(topic);
                    
                    monitorTopicDtos.add(monitorTopicDto);
                });
            } catch (Exception e) {
                log.error("get topicList by user", e);
            }
        }
        return PageUtil.toPage(monitorTopicDtos, null);
    }

    @Override
    public Set<String> getTopicList(String clusterId) throws InterruptedException, ExecutionException {
        return kafkaService.getKafkaUtil(clusterId).listTopics();
    }
}
